package org.lzfto.flink.demo.business.orders;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.concurrent.atomic.AtomicBoolean;

import org.apache.commons.lang3.ObjectUtils;
import org.lzfto.flink.demo.business.flinkTest.test.SimpleTestService;
import org.lzfto.flink.demo.business.flinkTest.test.TestUtil;
import org.lzfto.flink.demo.business.flinkTest.entity.TestInfo;
import org.lzfto.flink.demo.common.data.PageData;
import org.lzfto.flink.demo.common.data.PagingOperation;
import org.lzfto.flink.demo.common.result.PubResult;
import org.lzfto.flink.demo.config.AppConfiguration;
import org.lzfto.flink.demo.config.mqtt.CustomMqttMessage;
import org.lzfto.flink.demo.config.mqtt.MQTTConfig;
import org.lzfto.flink.demo.config.websocket.MyWebSocketHandler;
import org.lzfto.flink.demo.utils.SpringUtils;
import org.lzfto.flink.demo.utils.TimeUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Lazy;
import org.springframework.integration.mqtt.support.MqttHeaders;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.jdbc.core.BeanPropertyRowMapper;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.stereotype.Service;

import com.alibaba.fastjson2.JSON;

@Lazy
@Service
public class OrderService extends PagingOperation {
	private final Logger LOGGER = LoggerFactory.getLogger(getClass());
	
    private AtomicBoolean isSendingMQTT = new AtomicBoolean(false);
    private Date lastStartTime = null;
    
    @Lazy
    @Autowired
    private SimpleTestService testService;

    public PubResult<Integer> countTargetOrder(){
		String sql="select count(*) from orders_target";
		Integer totalQty=jdbcTemplate.queryForObject(sql, Integer.class);
		return PubResult.ok_data(totalQty);
	}

	public void deleteById(int id) {
	    String sql = "DELETE FROM orders WHERE id = ?";
	    jdbcTemplate.update(sql, id);
	}

	public PageData<Order> findAll(Order param, int pageNo, int pageSize) {
        String sql = "SELECT count(*) FROM orders WHERE 1=1 \n ";
        String pageSql = "SELECT * FROM orders WHERE 1=1 \n ";

        String whereSql = "";
        List<Object> args = new ArrayList<>();

        if (ObjectUtils.isNotEmpty(param.getCategoryCode())) {
            whereSql += " AND category_code = ?";
            args.add(param.getCategoryCode());
        }
        if (ObjectUtils.isNotEmpty(param.getBrand())) {
            whereSql += " AND brand  like  ?";
            args.add("%" + param.getBrand() + "%");
        }
        if (ObjectUtils.isNotEmpty(param.getZoneName())) {
            whereSql += " AND zone_name  like ?";
            args.add("%" + param.getZoneName() + "%");
        }

        // 处理分页参数
        pageSql = pageSql + "\n " + whereSql;
        sql += whereSql;
        return this.buildPage(pageNo, pageSize, sql, pageSql, args, Order.class);
    }
    
    public PageData<SimpleOrder> findAllSimple(Order param, int pageNo, int pageSize,boolean calTotal) {
        String sql = "SELECT count(*) FROM orders WHERE 1=1 \n ";
        String pageSql = "SELECT id, category_code, brand, price, age, sex, zone_name,\n"
        		+ "  cast(trim(replace(event_time,'UTC','')) as datetime)  as event_time \n"
        		+ "  FROM orders WHERE 1=1 \n ";

        String whereSql = "";
        List<Object> args = new ArrayList<>();

        if (ObjectUtils.isNotEmpty(param.getCategoryCode())) {
            whereSql += " AND category_code = ?";
            args.add(param.getCategoryCode());
        }
        if (ObjectUtils.isNotEmpty(param.getBrand())) {
            whereSql += " AND brand  like  ?";
            args.add("%" + param.getBrand() + "%");
        }
        if (ObjectUtils.isNotEmpty(param.getZoneName())) {
            whereSql += " AND zone_name  like ?";
            args.add("%" + param.getZoneName() + "%");
        }

        // 处理分页参数
        pageSql = pageSql + "\n " + whereSql;
        sql += whereSql;
        return this.buildPage(pageNo, pageSize, sql, pageSql, args, SimpleOrder.class,calTotal);
    }

    public Order findById(int id) {
        String sql = "SELECT * FROM orders WHERE id = ?";
        return jdbcTemplate.queryForObject(sql, new BeanPropertyRowMapper<>(Order.class), id);
    }

    public void insert(Order order) {
	    String sql = "INSERT INTO orders (id, event_time, order_id, product_id, category_id, "
	            + "category_code, brand, price, user_id, age, sex, zone_name) "
	            + "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)";
	    jdbcTemplate.update(sql, order.getId(), order.getEventTime(), order.getOrderId(), order.getProductId(),
	            order.getCategoryId(), order.getCategoryCode(), order.getBrand(), order.getPrice(), order.getUserId(),
	            order.getAge(), order.getSex(), order.getZoneName());
	}

	public PubResult<Boolean> releaseMqttLock(){
	    try{
	        this.isSendingMQTT.getAndSet(false);
	        return PubResult.ok("释放MQTT锁成功", true);
	    }
	    catch(Exception e){
	        return PubResult.ok("释放MQTT锁失败:"+e.getMessage(), false);
	    }
	}

	public void save(Order order) {
        if (order.getId() == null || order.getId() == 0) {
            insert(order);
        } else {
            update(order);
        }
    }

    /**
	 * 从数据库读取数据，并通过mqtt发送
	 * @param orderParam   订单参数
	 * @param pageQty      需要发送的页数
	 * @param pageSize     页面大小
	 * @param sendSocket   是否通过websocket发送消息
	 * @param interval     每次处理完成一行后所等待的时间(毫秒)
	 * @param testNo       测试编号      
	 * @return
	 */
	public PubResult<String> sendDataToMqtt(Order orderParam,int pageQty,int pageSize,int sendSocket,Integer interval,
			int testNo) {
		if (!AppConfiguration.getConf().getIsArtemisRegistered()) {
		    return PubResult.error("MQTT服务未注册");
		}
		
	    if (isSendingMQTT.get()) {
	        return PubResult.error("正在发送数据到MQTT，请稍后再试");
	    }
	
	    isSendingMQTT.getAndSet(true);
	    lastStartTime = new Date();
	    
	    TestInfo testInfo=testService.getTestInfo(testNo);
	    String webSocketClientName=testInfo.websocketClientName();
	    String mqttTopicSuffix=testInfo.mqttTopicSuffix();
	    String topic=TestUtil.buildMqttTopic(
	    		AppConfiguration.getConf().getAppProp().getMqttTopicPrefix(),mqttTopicSuffix);
	    
	    LOGGER.info("发送参数:"
	    		+ "\n pageQty={} "
	    		+ "\n pageSize={} "
	    		+ "\n sendSocket={} "
	    		+ "\n rowInterval={} "
	    		+ "\n testNo={} "
	    		+ "\n webSocketClient={}"
	    		+ "\n 接收者主题={}",
	    		pageQty,pageSize,sendSocket,interval,testNo,
	    		webSocketClientName,topic);
	    try {
	        MyWebSocketHandler.sendPubResultTo("开始通过MQTT发送订单数据(" + TimeUtil.formatDate(lastStartTime) + ")",webSocketClientName);
	    } catch (IOException e) {
	        e.printStackTrace();
	    }
	    // 创建一个线程用于发送数据
	    new Thread(() -> {
	    	int successCount = 0;
	    	int idx=0;
	    	int pauseInterVal=interval==null?0:interval.intValue();
	    	MessageChannel channel=SpringUtils.getBean(MQTTConfig.PRODUCER_CHANNEL_BEAN_NAME);
	    	
	    	//0.0 先删除目标库上的所有内容
			SimpleOrderService targetOrderService=SpringUtils.getBean(SimpleOrderService.class);
			targetOrderService.deleteAll();
	    	//1.0 发送表示开始的消息
	    	CustomMqttMessage<String> startMessage= CustomMqttMessage.buildStart();
	    	String startJson = JSON.toJSONString(startMessage);
	        Message<String> startMqttMessage = MessageBuilder.withPayload(startJson)
	        		.setHeader(MqttHeaders.TOPIC, topic)
	        		.build();
	        channel.send(startMqttMessage);
	        LOGGER.info("开始发送订单数据,接收者主题="+topic);
	        //2.0 发送消息体
	        for (int j = 0; j < pageQty; j++) {
	            PageData<SimpleOrder> pageData = findAllSimple(orderParam, j+1, pageSize,false);
	            LOGGER.info("获取第{}个分页数据",j+1);
	            List<SimpleOrder> orders = pageData.getRows();
	            if (orders==null || orders.size()==0) {
	                break;
	            }
	            //MessagingTemplate template=SpringUtils.getBean("artemisMessagingTemplate");
	            for (SimpleOrder order : orders) {
	                try {
	                    // 发送订单数据到MQTT
	                	idx++;
	                	CustomMqttMessage<SimpleOrder> dataMessage=new CustomMqttMessage<SimpleOrder>(CustomMqttMessage.MQTT_MESSAGE_TYPE_NORMAL,"订单",order);
	                    Message<String> dataMqttMessage = MessageBuilder.withPayload(JSON.toJSONString(dataMessage))
	                    		.setHeader(MqttHeaders.TOPIC, topic)
	                    		.build();
	                    boolean sent = channel.send(dataMqttMessage);
	                    if (sent) {
	                        successCount++;
	                        if (sendSocket==1) {
	                        	MyWebSocketHandler.sendPubResultTo("发送第" + successCount + "条数据 ❤",webSocketClientName);	
	                        }
	                    }
	                    // 休眠500毫秒
	                    // 发送太快，可能导致websocket断开，或者是Flink的DataStream无法处理。
	                    if (pauseInterVal>0) {
	                    	Thread.sleep(pauseInterVal);
	                    }
	                }catch (Exception e) {
	                    // 单条发送失败继续下一条
	                	try {
							MyWebSocketHandler.sendPubResultTo("发送第" + idx + "条数据失败!",webSocketClientName);
						} catch (IOException e1) {
						}
	                    continue;
	                }
	            }
	        }
	        
	        LOGGER.info("完成发送订单数据");
	        
	        //3.0 发送表示结束的消息
	        Message<String> endMqttMessage = MessageBuilder.withPayload(JSON.toJSONString(CustomMqttMessage.buildEnd()))
	        		.setHeader(MqttHeaders.TOPIC, topic)
	        		.build();
	        channel.send(endMqttMessage);
	        
	        //4.0发送WebSocket消息，告知客户端已经完成发送
	        try {
	        	PubResult<String> tmpResult=PubResult.ok_data("本次累计发送"+successCount+"条("+TimeUtil.formatDate(new Date())+")")
	        			.withStageType(PubResult.MESSAGE_TYPE_JOB_ISFINISHED);
				MyWebSocketHandler.sendTxtTo(JSON.toJSONString(tmpResult),webSocketClientName);
				LOGGER.info("发送订单到MQTT的任务已经完成，累计发送{}条",successCount);
			} catch (IOException e) {
				e.printStackTrace();
			}
	        isSendingMQTT.set(false);
	        
	    }).start();
	
	    return PubResult.ok("已经启动MQTT发送任务");
	
	}

	public void update(Order order) {
        String sql = "UPDATE orders SET  " + " brand=?, price=?,  zone_name=? WHERE id=?";
        jdbcTemplate.update(sql, order.getBrand(), order.getPrice(), order.getZoneName(), order.getId());
    }
}
